package org.gjy.m8.thread;

import org.junit.Test;

import java.time.LocalDateTime;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @author 宫静雨
 * @version 1.0
 * @since 2023-09-07 08:42:00
 */
public class Cf2Test {

    @Test
    public void rpc() {
        String s = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("rpc1");
            return "rpc1";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("rpc2");
            return "rpc2";
        }), (l, r) -> l + "&" + r).join();
        System.out.println(s);
    }

    @Test
    public void drink() {
        // 任务一: 洗水壶, 烧开水
        CompletableFuture<Boolean> hotJob = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("洗好水壶, 烧开水");
                TimeUnit.SECONDS.sleep(2);
                System.out.println("水开了");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return true;
        });

        // 任务二: 洗茶壶、洗茶杯和拿茶叶
        CompletableFuture<Boolean> washJob = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("洗茶杯");
                TimeUnit.SECONDS.sleep(2);
                System.out.println("洗完了");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return true;
        });

        // 任务三: 负责泡茶
        String s = hotJob.thenCombine(washJob, (h, w) -> {
            if (h && w) {
                System.out.println("泡茶喝, 茶喝完");
                return "OK";
            }
            return "FF";
        }).join();
        System.out.println(s);

    }

    @Test
    public void acceptEither() {
        CompletableFuture.supplyAsync(() -> {
            try {
                int timeout = ThreadLocalRandom.current().nextInt(5);
                System.out.println("A = " + timeout);
                TimeUnit.SECONDS.sleep(timeout);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("A");
            return "A";
        }).acceptEither(CompletableFuture.supplyAsync(() -> {
            try {
                int timeout = ThreadLocalRandom.current().nextInt(5);
                System.out.println("B = " + timeout);
                TimeUnit.SECONDS.sleep(timeout);
                System.out.println("B");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            return "B";
        }), v -> System.out.println("acceptEither===" + v)).join();
    }

    @Test
    public void runAfterEither() {
        CompletableFuture.supplyAsync(() -> {
            try {
                int timeout = ThreadLocalRandom.current().nextInt(5);
                System.out.println("A = " + timeout);
                TimeUnit.SECONDS.sleep(timeout);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("A");
            return "A";
        }).runAfterEither(CompletableFuture.supplyAsync(() -> {
            try {
                int timeout = ThreadLocalRandom.current().nextInt(5);
                System.out.println("B = " + timeout);
                TimeUnit.SECONDS.sleep(timeout);
                System.out.println("B");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            return "B";
        }), () -> System.out.println("runAfterEither")).join();
    }

    @Test
    public void applyToEither() {
        String s = CompletableFuture.supplyAsync(() -> {
            try {
                int timeout = ThreadLocalRandom.current().nextInt(5);
                System.out.println("A = " + timeout);
                TimeUnit.SECONDS.sleep(timeout);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("A");
            return "A";
        }).applyToEither(CompletableFuture.supplyAsync(() -> {
            try {
                int timeout = ThreadLocalRandom.current().nextInt(5);
                System.out.println("B = " + timeout);
                TimeUnit.SECONDS.sleep(timeout);
                System.out.println("B");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            return "B";
        }), v -> v).join();
        System.out.println("s = " + s);
    }

    @Test
    public void allOf() {
        CompletableFuture.allOf(
                CompletableFuture.supplyAsync(() -> {
                    System.out.println("10");
                    return 10;
                }),
                CompletableFuture.supplyAsync(() -> {
                    System.out.println("20");
                    return 20;
                }), CompletableFuture.runAsync(() -> {
                    System.out.println("30");
                })
        ).join();
    }

    @Test
    public void thenAcceptBoth() {
        CompletableFuture.supplyAsync(() -> 10)
                .thenAcceptBoth(CompletableFuture.supplyAsync(() -> 20),
                        (l, r) -> {
                            System.out.println("l = " + l);
                            System.out.println("r = " + r);
                        })
                .join();
    }

    @Test
    public void runAfterBoth() {
        CompletableFuture.supplyAsync(() -> 10)
                .runAfterBoth(CompletableFuture.supplyAsync(() -> 20),
                        () -> System.out.println("runAfterBoth"))
                .join();
    }

    @Test
    public void thenCombine() {
        Integer integer = CompletableFuture.supplyAsync(() -> 10)
                .thenCombine(CompletableFuture.supplyAsync(() -> 20),
                        (l, r) -> {
                            System.out.println("l = " + l);
                            System.out.println("r = " + r);
                            return l + r;
                        })
                .join();
        System.out.println("integer = " + integer);
    }

    @Test
    public void thenCompose() {
        String s = CompletableFuture.supplyAsync(() -> "Hello")
                .thenCompose(v -> CompletableFuture.supplyAsync(() -> v + "==" + v))
                .join();
        System.out.println(s);
    }

    @Test
    public void thenAccept() {
        CompletableFuture.supplyAsync(() -> "Hello")
                .thenAccept((v) -> System.out.println(v)).join();
    }

    @Test
    public void thenRun() {
        CompletableFuture.supplyAsync(() -> "Hello")
                .thenRun(() -> System.out.println("thenRun")).join();
    }

    @Test
    public void thenApply() {
        Integer integer = CompletableFuture.supplyAsync(() -> {
                    int i = ThreadLocalRandom.current().nextInt(10);
                    System.out.println("producer: " + i);
                    return i;
                })
                .thenApply(n -> {
                    if (n >= 5) {
                        return n * 2;
                    } else {
                        return 0;
                    }
                }).join();
        System.out.println("integer = " + integer);
    }

    @Test
    public void pool() {
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                processors,
                processors * 2,
                1, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(processors * 10),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        String s = CompletableFuture.supplyAsync(() -> "Hello", pool)
                .join();
        System.out.println("s = " + s);
    }

    @Test
    public void handle() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        CompletableFuture<String> f = future.handle((v, t) -> {
            if (Objects.nonNull(t)) {
                System.out.println("发生了异常");
                return "ERROR";
            } else {
                System.out.println("v = " + v);
                return "SUCCESS";
            }
        });
        System.out.println(future.join());
        System.out.println(f.join());
    }

    @Test
    public void hook() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello CompletableFuture");
        future.whenComplete((v, e) -> {
            System.out.println("whenComplete v = " + v);
            System.out.println("whenComplete e = " + e.getMessage());
        });
        future.exceptionally(t -> {
            System.out.println("exceptionally: " + t.getMessage());
            return t.getMessage();
        });
        String s = future.join();
        System.out.println("s = " + s);
    }

    @Test
    public void hook2() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello CompletableFuture");

        future.exceptionally(t -> {
            System.out.println("exceptionally: " + t.getMessage());
            return t.getMessage();
        });
        future.cancel(true);
        String s = future.join();
        System.out.println(s);
    }

    @Test
    public void runAsync() {
        // join 的含义可参考上文
        CompletableFuture.runAsync(() -> System.out.println(LocalDateTime.now()))
                .join();
    }

    @Test
    public void runAsyncExecutors() {
        // join 的含义可参考上文
        ExecutorService pool = Executors.newFixedThreadPool(1);
        CompletableFuture.runAsync(() -> System.out.println(Thread.currentThread().getName()
                        + "=" + LocalDateTime.now()), pool)
                .join();
    }

    @Test
    public void supplyAsync() {
        // join 的含义可参考上文
        LocalDateTime time = CompletableFuture.supplyAsync(LocalDateTime::now)
                .join();
        System.out.println("time = " + time);
    }

    @Test
    public void supplyAsyncExecutors() {
        // join 的含义可参考上文
        ExecutorService pool = Executors.newFixedThreadPool(1);
        LocalDateTime time = CompletableFuture.supplyAsync(LocalDateTime::now, pool)
                .join();
        System.out.println("time = " + time);
    }

    @Test
    public void test1() throws InterruptedException {
        CompletableFuture.supplyAsync(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return "Hello";
                })
                .thenApply(x -> x + "==" + x)
                .thenAccept(x -> System.out.println(x))
                .thenRun(() -> System.out.println("END"))
                .join();
    }

}
